package dir.simulation;

import java.util.HashMap;
import java.util.PriorityQueue;
import java.util.Random;
import java.util.Vector;

import dht.data.Access;
import dht.data.AccessType;
import dht.data.Message;
import dht.data.MessageUtil;
import dht.simulation.DhtException;
import dht.simulation.GlobalVariables;
import dht.statistics.DhtStatistics;
import dse.modules.eln.EarthLikeNetwork;
import dse.modules.eln.EarthLikeNetworkCostCalculator;

public class DirFs {
	protected DirMasterNode masterNode;
	protected HashMap<Integer, DirDataNode> dataNodes;
	protected EarthLikeNetwork network;
	protected EarthLikeNetworkCostCalculator networkCostCalculator;
	protected HashMap<Integer, Access> pendingAccesses;
	protected PriorityQueue<Message> messageQueue;
	protected Random ran;
	public int totalAccess;
	public int totalAccessProcessed;
	public int systemAccessId;

	public DirFs(EarthLikeNetwork network) {
		this.masterNode = new DirMasterNode(GlobalVariables.masterId,
				network.maxNodeId());
		this.dataNodes = new HashMap<Integer, DirDataNode>();
		this.network = network;
		this.networkCostCalculator = new EarthLikeNetworkCostCalculator();
		this.pendingAccesses = new HashMap<Integer, Access>();
		this.messageQueue = new PriorityQueue<Message>();
		this.ran = new Random(GlobalVariables.seed);
		this.totalAccess = 0;
		this.totalAccessProcessed = 0;
		this.systemAccessId = 0;
		initialize();
	}

	private void initialize() {
		for (int i = 1; i <= network.maxNodeId(); ++i) {
			DirDataNode newNode = new DirDataNode(i, masterNode.masterId);
			dataNodes.put(i, newNode);
		}
	}

	public void addAccess(Access access) throws DhtException {
		Message msg = null;
		if (access.accessType == AccessType.READ
				|| access.accessType == AccessType.WRITE) {
			DirDataNode fromNode = dataNodes.get(access.requesterId);
			msg = fromNode.clientStartAccess(access);
		}
		if (msg == null) {
			totalAccess++;
			totalAccessProcessed++;
			DhtStatistics.totalAccessCount++;
			DhtStatistics.totalAccessProcessedCount++;
		} else {
			totalAccess++;
			DhtStatistics.totalAccessCount++;
			access.msgSent = 1;
			pendingAccesses.put(access.accessId, access);
			sendMsg(msg);
			messageQueue.add(msg);
		}
	}

	public void run(double toTime, boolean additional) throws DhtException {

		while (!messageQueue.isEmpty()) {
			Message msg = messageQueue.peek();
			if (msg.getCurrentTime() >= toTime) {
				break;
			}
			messageQueue.poll();
			Access access = pendingAccesses.get(msg.accessId);
			DirDataNode toNode = dataNodes.get(msg.toId);
			Vector<Message> msgList = null;
			if (MessageUtil.isDirClient2MasterMsg(msg)) {
				msgList = masterNode.masterProcessMsg(msg, access);
			} else if (MessageUtil.isDirMaster2ClientMsg(msg)) {
				msgList = toNode.clientProcessMsg(msg, access);
			} else if (MessageUtil.isDirClient2ServerMsg(msg)) {
				msgList = toNode.serverProcessMsg(msg, access);
			} else if (MessageUtil.isDirServer2ClientMsg(msg)) {
				msgList = toNode.clientProcessMsg(msg, access);
			} else {
				throw new DhtException("msgType unknown");
			}
			access.messageFlow.add(msg);
			if (msgList != null) {
				for (Message newMsg : msgList) {
					sendMsg(newMsg);
					access.msgSent++;
					messageQueue.add(newMsg);
				}
				// pendingAccesses.put(access.accessId, access);
			} else if (access.messageFlow.size() == access.msgSent) {
				pendingAccesses.remove(access.accessId);
				totalAccessProcessed++;
				DhtStatistics.accessReport(access);
				DhtStatistics.totalAccessProcessedCount++;
				if (!access.valid()) {
					System.err.println("access is invalid");
					access.dump();
					throw new DhtException("access is invalid");
				}
			}
		}
		System.out.println("pendingAccesses count after period: "
				+ pendingAccesses.size());
		System.out.println("messageQueue size after period: "
				+ messageQueue.size());
	}

	public void sendMsg(Message msg) {
		msg.networkCost = networkCostCalculator.cost(network,
				network.lookupNode(msg.fromId), network.lookupNode(msg.toId),
				msg.msgSize);
	}
}
